package com.bleeth.flow.core.common;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import com.jd.platform.async.callback.ICallback;
import com.jd.platform.async.callback.IWorker;
import com.jd.platform.async.worker.WorkResult;
import com.jd.platform.async.wrapper.WorkerWrapper;
import lombok.Data;
import org.apache.log4j.Logger;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

import java.util.List;
import java.util.Map;

/**
 * @author ：Bleeth
 * @date ：2021-08-05
 * @description：插件抽象类
 */

@Data
public abstract class APlugin implements IWorker<Dataset<Row>, Dataset<Row>>, ICallback<Dataset<Row>, Dataset<Row>> {

    private Logger logger = Logger.getLogger(getClass());


    /**
     * 插件ID
     */
    protected String id;


    /**
     * 插件NAME
     */
    protected String name;

    /**
     * 插件描述
     */
    protected String description;
    /**
     * 插件输出组件
     */
    protected List<String> nextList;


    /**
     * 插件输入组件
     */
    protected List<String> fromList;

    /**
     * 插件输出组件
     */
    protected String type;

    /**
     * 调试
     */
    protected boolean debug;


    /**
     * 初始化一些环境，大部分插件不需要设置，
     * 特殊的比如hive,需要设置hive-site.xml
     */
    public void init() {

    }


    @Override
    public void result(boolean success, Dataset<Row> input, WorkResult<Dataset<Row>> workResult) {
       if(input!=null){
           logger.info(StrUtil.format("执行完成：{},处理数据量：{}", this.name, input.count()));
       }
    }

    @Override
    public Dataset<Row> action(Dataset<Row> input, Map<String, WorkerWrapper> allWrappers) {
        if (CollUtil.isNotEmpty(fromList)) {
            fromList.stream().forEach(from -> {
                logger.info("from -> this:" + from + "---->" + this.name + "----->");
            });
        }
        return null;
    }

    public JSONObject toPluginString() {
        JSONObject jsonObject = new JSONObject(this);
        return jsonObject;
    }

}
